/***************************************************************************
 *   Copyright (C) 2009 by Bernhard Neuhofer   *
 *   mail@bneu.at   *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program is distributed in the hope that it will be useful,       *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 *   GNU General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, write to the                         *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/

using namespace std;
#include "../tools/log.h"
#include "AbstractLp.h"
#include "inQueue.h"
#include "vectorOutQueue.h"
#include "memento.h"
#include "sqtwAdapter.h"
#include "rollbackableAdapter.h"

#include <iostream>
#include <vector>
#include <stack>
#include <map>
#include <limits>
#include <typeinfo>


#include "rollbackableLp.h"

RollbackableLp::State::State ( double lvt,vector<InQueue*>inQueues,VectorOutQueue* outQueue,Memento* memento )
{
	_lvt=lvt;
	_memento=memento;
	_storedOutQueue=outQueue;
	_lastReads.clear();
	for ( int i=0; i<inQueues.size();i++ )
		_lastReads[i]=inQueues[i]->getLastRead();
}

map<unsigned int,Message*> RollbackableLp::State::getLastReads()
{
	return _lastReads;
}

double RollbackableLp::State::getLvt()
{
	return _lvt;
}

VectorOutQueue* RollbackableLp::State::getOutQueue()
{
	return _storedOutQueue;
}

Memento* RollbackableLp::State::getMemento()
{
	return _memento;
}


void RollbackableLp::doStep()
{
	FILE_LOG(logDEBUG)<<"rollbackableLp::dostep"<<endl;
	//Update Output collector
	getSmOutput()->incSteps();
	FILE_LOG(logDEBUG)<<"rollbackableLp::dostep2:"<<getGvtStats()->getColor()<<endl;
	//Snapshot the Sms relevant gvtStats
	getStepStats()->setColor(true);
	//getStepStats()->setColor ( getGvtStats()->getColor() );
	FILE_LOG(logDEBUG)<<"rollbackableLp::dostep after setcolor"<<endl;
	getStepStats()->setTmin ( numeric_limits<double>::max() );

	getStepStats()->setMd ( true,getGvtStats()->getMd ( true ) );
	getStepStats()->setMd ( false,getGvtStats()->getMd ( false ) );

	FILE_LOG(logDEBUG)<<endl<<"========> doStep of LP    "<<getId() <<" called"<<endl;
	FILE_LOG(logDEBUG)<<"LVT: "<<getLvt() <<" | Color: "<<getStepStats()->getColor()
	<<" | mdw: "<<getStepStats()->getMd ( true ) <<" | mdr: "<<getStepStats()->getMd ( false )
	<<" | lvtSm: "<<getGvtStats()->getLvtSm() <<" | tmin: "<<getStepStats()->getTmin() <<endl;
	FILE_LOG(logDEBUG)<< "---------------------------------------------------------------------------------------------"
	<<endl;

	FILE_LOG(logDEBUG)<<"Queue 0: "<<_inQueues[0]<<endl;
	if ( _inQueues.size() >1 ) cout<<"Queue 1: "<<_inQueues[1]<<endl;
	FILE_LOG(logDEBUG)<<"Unprocessed message in InBuffer:"<<getAssocMh()->bufferSize ( getId() ) <<endl<<endl;
	cout
	<<"---------------------------------------------------------------------------------------------"
	<<endl;
	FILE_LOG(logDEBUG)<<"rollbackableLp::dostep3"<<endl;
	/* Copys all messages from the InBuffer to the appropriate InQueues.
	* Rollbacks to an consistant state if a straggler message is received.
	* After the call the InBuffer is empty and the state is consistent. */
	processInBuffer();
	FILE_LOG(logDEBUG)<<"rollbackableLP: after processInBuffer"<<endl;
	unsigned int srcIndex=0;
	if ( dynamic_cast<SqtwAdapter*>(getAssocAd())!=NULL )
	{
	FILE_LOG(logDEBUG)<<"rollbackableLP: Sqtwqueue selected1"<<endl;
		srcIndex= ( ( SqtwAdapter* ) getAssocAd() )->outOfWhichQueue ( getAssocMe() );
	FILE_LOG(logDEBUG)<<"rollbackableLP: Sqtwqueue selected2"<<endl;
	}
	
	/* If src-Inqueue exists and contains unread messages
	 then save the state an process the messages*/
	FILE_LOG(logDEBUG)<<"rollbackableLP: inqueues.size:"<<_inQueues.size()<<"| srcIndex:"<<srcIndex<<endl;
	if ( _inQueues.size() >srcIndex )
	{
		VectorInQueue* srcQueue= ( VectorInQueue* ) _inQueues[srcIndex];
		FILE_LOG(logDEBUG)<<"rollbackableLP:scrQueue:"<<srcQueue->size()<<endl;
FILE_LOG(logDEBUG)<<"rollbackableLP:unread messages?:"<<srcQueue->containsUnread()<<endl;
		if ( srcQueue->containsUnread() )
		{
			FILE_LOG(logDEBUG)<<"rollbackableLP:unread messages2?:"<<srcQueue->containsUnread()<<endl;
			_terminated=false;
			//Tailor state object and push it to the stateStack
			double lvt=getLvt();
			_stateStack.push_back (
			    new State (
			        lvt,_inQueues,_outQueue,
			        ( ( RollbackableAdapter* ) getAssocAd() )->createMemento ( getAssocMe() ) )
			);
			FILE_LOG(logDEBUG)<<"LP "<<getId() <<" successfully saved its state for LVT "<<lvt<<endl;

			//Flushes the OutQueue vector
			_outQueue->clear();

			//Call Me to process nextUnread
			FILE_LOG(logDEBUG)<<"LP "<<getId() <<" fetched message from queue #"<<srcIndex<<" and passed it to ME:";
			srcQueue->getNextUnread()->interpret ( ( Interpreter* ) getAssocMe() );
		}
		/* If src-Inqueue is empty switch to terminated state */
		else
		{
			FILE_LOG(logDEBUG)<<"rollbackableLP:noops inc"<<endl;
			//Update Output collector
			getSmOutput()->incNoops();
			_terminated=true;
		}
	}

	FILE_LOG(logDEBUG)<<endl<<"Last GVT approximation: "<<getGvtStats()->getGvt() <<endl;
	FILE_LOG(logDEBUG)<<"rollbackableLP: getlvt():"<<getLvt()<<"| "<<getGvtStats()->getGvt()<<endl;
	if ( getLvt() <getGvtStats()->getGvt() )
	{
		FILE_LOG(logDEBUG)<<"LP "<<getId() <<" LVT ("<<getLvt() <<") < GVT ("<<getGvtStats()->getGvt() <<") approximation!!"<<endl;
		FILE_LOG(logDEBUG)<<"LVT: "<<getLvt() <<" | Color: "<<getStepStats()->getColor()
		<<" |mdw: "<<getStepStats()->getMd ( true ) <<" | mdr: "<<getStepStats()->getMd ( false )
		<<" | lvtSm: "<<getGvtStats()->getLvtSm()
		<<" | tmin: "<<getStepStats()->getTmin() <<endl;
	}

	//Report changes back to SM
	getGvtStats()->setMd ( true,getStepStats()->getMd ( true ) );
	getGvtStats()->setMd ( false,getStepStats()->getMd ( false ) );
	
	// fossil collect: prune queues, empty state-stack if new GVT available
	FILE_LOG(logDEBUG)<<"rollbackableLP: getlvt():"<<getGvtStats()->getGvt()<<"| "<<getStepStats()->getGvt()<<endl;
	if ( getGvtStats()->getGvt() >getStepStats()->getGvt() )
	{
		getStepStats()->setGvt ( getGvtStats()->getGvt() );
		FILE_LOG(logDEBUG)<<"Calling fossilCollect: "<<endl;
		fossilCollect();

	}
FILE_LOG(logDEBUG)<<"rollbackableLP: finished dostep()"<<endl;
}

double RollbackableLp::getLvt()
{
	double lvt=numeric_limits<double>::max();
	double ct;
	if ( !_terminated )
	{
		lvt=-1.0;
		vector<InQueue*>::iterator it;
		for ( it=_inQueues.begin();it!=_inQueues.end();++it )
		{
			ct= ( *it )->getChannelTime();
			if ( ct>lvt ) lvt=ct;
		}
	}
	return lvt;
}
void RollbackableLp::processInBuffer()
{
	unsigned int destIndex=0;
	unsigned int missing=0;
	FILE_LOG(logDEBUG)<<"RollbackableLp::Process in Buffer"<<endl;
	// For every unprocessed message in the InBuffer
	while ( !getAssocMh()->bufferEmpty ( getId() ) )
	{
		destIndex=0;
		FILE_LOG(logDEBUG)<<"RollbackableLp:Process MessageBuffer2:"<<getId()<<endl;
		//Get the message from the InBuffer
		Message* message=getAssocMh()->readBuffer ( getId() );
		FILE_LOG(logDEBUG)<<"RollbackableLp:read message from messagehandler"<<endl;
		//GVT algo: log received message
		getStepStats()->decMd ( message->getColor() );
		FILE_LOG(logDEBUG)<<"rollbackableLp:Decreased md counter for color: "<<message->getColor() <<endl;

		//For SQTW adapters ask the adapter for the right queue
		if ( dynamic_cast<SqtwAdapter*>( getAssocAd() ) !=NULL )
		{
			
			destIndex= ( ( SqtwAdapter* ) getAssocAd() )->intoWhichQueue ( message );
			FILE_LOG(logDEBUG)<<"RollbackableLp:SQTW Adapter found. Queue selected:"<<destIndex<<endl;
		}

		// If it is the first message for this queue, create the queue and all queues up to this on
		if ( _inQueues.size() <=destIndex )
		{
			FILE_LOG(logDEBUG)<<"RollbackableLp: Creating missing InQueue!"<<endl;
			missing=destIndex-_inQueues.size() +1;
			for ( int i=0;i<missing;i++ )
				_inQueues.push_back ( new VectorInQueue );
		}

		//Fetch the reference to the right InQueue
		VectorInQueue* destQueue= ( VectorInQueue* ) _inQueues[destIndex];

		Message* dual=message->createDual();

		//Check for straggler. Rollback may be necessary if timestamp of message <= channel time
		if (
		    ( message->getTimestamp() <destQueue->getChannelTime() && message->isPositive() && !destQueue->contains ( dual ) )
		    ||
		    ( message->getTimestamp() <=destQueue->getChannelTime() &&!message->isPositive() && destQueue->contains ( dual ) )
		)
		{
			FILE_LOG(logDEBUG)<<"rollbackablelp:Update Output collector"<<endl;
			//Update Output collector
			getSmOutput()->incRollbacks();

			FILE_LOG(logDEBUG)<<"## LP "<<getId() <<" message ["<<message->getTimestamp() <<"|"
			<< ( message->isPositive() ?"+":"-" ) <<"|"<< ( message->getColor() ?"w":"r" )
			<<"] for queue #"<<destIndex<<" is a straggler message!"<<endl;
			State* oldState=_stateStack.back();
			_stateStack.pop_back();

			//empty current OutQueue first
			while( !_outQueue->isEmpty())
			{
				Message* msg=_outQueue->first();
				FILE_LOG(logDEBUG)<<"## LP "<<getId()<<" generates antimessage:"<<endl;
				
				Message* anti=msg->createDual();
				anti->setColor(getStepStats()->getColor());
				getAssocMh()->sendMessage(anti);
				_outQueue->remove(msg);
	
				//Update Output collector
				getSmOutput()->incAntiMessages();

				// GVT algo: log sent message
				getStepStats()->incMd(getStepStats()->getColor());
				if(anti->getTimestamp()<getStepStats()->getTmin())
					getStepStats()->setTmin(anti->getTimestamp());
			}
			//Rollback to the earlisest state before the stragglers timestamp
			if (message->isPositive())
				while (oldState->getLvt()>message->getTimestamp())
				{
					// send antimessage
					while (!oldState->getOutQueue()->isEmpty())
					{
						Message* msg=oldState->getOutQueue()->first();
						FILE_LOG(logDEBUG)<<"## LP "<<getId()<<" generates antimessage:"<<endl;
						
						//send directly, don't put copy into OutQueue
						Message* anti=msg->createDual();
						anti->setColor(getStepStats()->getColor());

						getAssocMh()->sendMessage(anti);
						oldState->getOutQueue()->remove(msg);

						//Update Output collector
						getSmOutput()->incAntiMessages();
			
						//GVT algo: log sent message
						getStepStats()->incMd(getStepStats()->getColor());
				
						if (anti->getTimestamp()<getStepStats()->getTmin())
							getStepStats()->setTmin(anti->getTimestamp());
					}
					oldState=_stateStack.back();
					_stateStack.pop_back();
				}

		
			else
			{
				while( oldState->getLvt()>=message->getTimestamp() ){
					//send antimessages
					while( !oldState->getOutQueue()->isEmpty() ){
						Message* msg=oldState->getOutQueue()->first();
						FILE_LOG(logDEBUG)<<"## LP "<<getId()<<" generates antimessage:"<<endl;
	
						//send directly, don't put copy into OutQueue
						Message* anti=msg->createDual();
						anti->setColor(getStepStats()->getColor());
						getAssocMh()->sendMessage(anti);
						oldState->getOutQueue()->remove(msg);
			
						//Update Output collector
						getSmOutput()->incAntiMessages();
			
						//GVT algo: log sent message			
						getStepStats()->incMd(getStepStats()->getColor());

						if(anti->getTimestamp()<getStepStats()->getTmin())
							getStepStats()->setTmin(anti->getTimestamp());
					}
				oldState=_stateStack.back();
				_stateStack.pop_back();
				}
			}
		
			FILE_LOG(logDEBUG)<<"## LP " <<getId()<<" is rolling back to state with LVT "<<oldState->getLvt()<<endl;
	
			//Restore old state
			_outQueue=oldState->getOutQueue();
			((RollbackableAdapter*)getAssocAd())->setMemento(getAssocMe(),oldState->getMemento());
			map<unsigned int, Message*> oldLastReads=oldState->getLastReads();
			for (int i=0;i<_inQueues.size();i++)
			  _inQueues[i]->setLastRead(oldLastReads[i]);
		} //end lcc check and roleback

		FILE_LOG(logDEBUG)<<"rollbackablelp:Annihilate dual-message if present or put the message into the destination queue	"<<endl;
		// Annihilate dual-message if present or put the message into the destination queue	
		if (destQueue->contains(dual))
			{	
				destQueue->remove(dual);
				FILE_LOG(logDEBUG)<<"LP "<<getId()<<" annihilated dual message from ["<<dual->getTimestamp()<<"|"
					<<(message->isPositive()?"+":"-")<<"|"
					<<(message->getColor()?"w":"r")<<"] in queue #"<<destIndex<<endl;
			}else
			{
				//Update Output collector
				getSmOutput()->incReceived();
				destQueue->insert(message);
				FILE_LOG(logDEBUG)<<"LP "<<getId()<<" queued message ["<< message->getTimestamp()<<"|"
					<<(message->isPositive()?"+":"-")<<"|"
					<<(message->getColor()?"w":"r")<<"] into queue #"<<destIndex<<endl;
			}
		

	}//end while (getAssocMh...)
}
void RollbackableLp::sendMessage ( Message* message )
{
	/* Color the message in the stepColor */
	message->setColor ( getStepStats()->getColor() );

	/*Send via Mh */
	getAssocMh()->sendMessage ( message );

	/*Store the sent message in _OutQueue */
	_outQueue->insert ( message );

	/* Update Output collector */
	getSmOutput()->incSent();

	/* GVT algo: log sent message */
	getStepStats()->incMd ( getStepStats()->getColor() );

	if ( message->getTimestamp() <getStepStats()->getTmin() )
		getStepStats()->setTmin ( message->getTimestamp() );
}

void RollbackableLp::fossilCollect()
{
	if ( _stateStack.size() >2 )
	{
		// Prune StateStack
		while ( _stateStack[1]->getLvt() <getStepStats()->getGvt() && _stateStack.size() >2 )
		{
			_stateStack.pop_front();
		}
	}
	//Prune InQieies according to last possible rollbackable state on the stack
	for ( int i=0;i <_inQueues.size();i++ )
	{
		if ( _stateStack[0]->getLastReads() [i]!=NULL )
		{
			_inQueues[i]->fossilCollect ( _stateStack[0]->getLastReads() [i]->getTimestamp() );
		}
	}
}
